package xubo.wangcaifeng.love.Utils

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row
import scalikejdbc.config.DBs
import scalikejdbc.{DB, SQL}

import scala.io.Source

object SaveData {

  def apply(result: RDD[((String, String), List[Int])]) = {
    DBs.setupAll()
    DB.localTx(implicit session =>{
        result.foreach(t=>{
          SQL("replace into localCount(proandcity,requestmode,processnode,iseffective,isbilling,isbid,iswin,adorderid) values(?,?,?,?,?,?,?,?)")
          .bind(t._1,t._2(0),t._2(1),t._2(2),t._2(3),t._2(4),t._2(5),t._2(6))
          .update()
          .apply()
      })
    })
  }


  def apply(t:Row): List[Double] = {
    val requestmode = t.getAs[Int]("requestmode")
    //获取流程节点
    val processnode = t.getAs[Int]("processnode")
    //获取有效标识
    val iseffective = t.getAs[Int]("iseffective")
    //获取是否收费的字段
    val isbilling = t.getAs[Int]("isbilling")
    //获取是否rtb
    val isbid = t.getAs[Int]("isbid")
    //获取是否竞价成功的字段
    val iswin = t.getAs[Int]("iswin")
    //获取广告id的字段
    val adorderid = t.getAs[Int]("adorderid")
    //获取winprice
    val winprice = t.getAs[Double]("winprice")
    //获取adpayment
    val adpayment = t.getAs[Double]("adpayment")
    List[Double](
      if (requestmode == 1 && processnode >= 1) 1 else 0,
      if (requestmode == 1 && processnode >= 2) 1 else 0,
      if (requestmode == 1 && processnode == 3) 1 else 0,
      if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) 1 else 0,
      if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0,
      if (requestmode == 2 && iseffective == 1) 1 else 0,
      if (requestmode == 3 && iseffective == 1) 1 else 0,
      if (iseffective == 1 && isbilling == 1 && iswin == 1) winprice/1000 else 0.0,
      if (iseffective == 1 && isbilling == 1 && iswin == 1) adpayment/1000 else 0.0)
  }
}
